package com.binance.notification.web.service.impl;

import java.io.IOException;
import java.io.InputStream;
import java.text.MessageFormat;
import java.util.Date;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Scanner;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.env.Environment;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.binance.notification.api.push.PushCommand;
import com.binance.notification.api.push.notification.AbstractNotification;
import com.binance.notification.api.push.notification.AndroidPayload;
import com.binance.notification.api.push.notification.AppPayload;
import com.binance.notification.api.push.notification.IosPayload;
import com.binance.notification.api.push.notification.MultipleNotificationPacket;
import com.binance.notification.api.push.notification.PcPayload;
import com.binance.notification.api.push.notification.SimpleNotification;
import com.binance.notification.api.push.notification.SingleNotificationPacket;
import com.binance.notification.api.push.notification.WebPayload;
import com.binance.notification.web.push.JClient;
import com.binance.notification.web.push.PushResult;
import com.binance.notification.web.push.RequestCallback;
import com.binance.notification.web.push.notification.PcNotification;
import com.binance.notification.web.push.notification.WebNotification;
import com.binance.notification.web.service.IDeviceService;
import com.binance.notification.web.service.IPushService;
import com.binance.notification.web.utils.Constants;
import com.binance.notification.web.utils.Worker;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Queues;
import com.google.common.collect.Sets;

import cn.jiguang.common.DeviceType;
import cn.jiguang.common.resp.APIConnectionException;
import cn.jiguang.common.resp.APIRequestException;
import cn.jpush.api.push.model.Platform;
import cn.jpush.api.push.model.PushPayload;
import cn.jpush.api.push.model.audience.Audience;
import cn.jpush.api.push.model.audience.AudienceTarget;
import cn.jpush.api.push.model.notification.AndroidNotification;
import cn.jpush.api.push.model.notification.IosAlert;
import cn.jpush.api.push.model.notification.IosNotification;
import cn.jpush.api.push.model.notification.Notification;
import io.jsonwebtoken.lang.Collections;
import lombok.extern.slf4j.Slf4j;

/**
 * 推送逻辑的总的实现类，其内部实现了推送的调度方式，会调用具体的推送实现类进行推送
 *
 * @author louis
 * @date 2018-04-18 10:13:02
 */
@Component
@Slf4j
public class PushServiceImpl implements IPushService {

    @Autowired
    private JClient jClient;
    @Resource
    private RedisTemplate<String, String> redisTemplate;
    @Value("${jpush.android.enabled:false}")
    private boolean jpushAndroidEnabled;
    @Value("${jpush.ios.enabled:false}")
    private boolean jpushIosEnabled;
    @Value("${jpush.pc.enabled:false}")
    private boolean jpushPcEnabled;
    @Value("${jpush.web.enabled:false}")
    private boolean jpushWebEnabled;
    @Value("${jpush.batch.count:1000}")
    private int batchThreshold;
    @Value("${jpush.future.task.count:1}")
    private int futureTaskCount;
    @Value("${jpush.use.i18n:false}")
    private boolean isUseI18n;
    @Value("${jpush.ios.badge.enabled:true}")
    private boolean iosBadgeEnabled;
    @Value("${jpush.ios.badge.value:1}")
    private int iosBadgeValue;
    @Resource
    private Environment env;
    private boolean production;
    @Autowired
    private IDeviceService deviceService;
    @Autowired
    private Worker worker;

    private final Map<String, Map<String, String>> i18nResources = Maps.newHashMap();
    private static final LinkedBlockingQueue<String> QUEUE = new LinkedBlockingQueue<>();
    private static final ThreadLocal<Map<String, List<String>>> threadLocalMap = new ThreadLocal<>();
    private static final ThreadLocal<Map<PushCommand, List<String>>> threadLocal = new ThreadLocal<>();

    // 发送一条通知一次可以发送的别名的最大个数
    private static final int aliasThreshold = 1000;

    // 从队列中获取PushPayload，凑够1000个就一次性发送；如果50毫秒内没有获取到PushPayload，则直接发送已收集到的PushPayload
    private void sendFromQueue() {
        List<String> payloads = Lists.newArrayList();
        for (;;) {
            String payload = null;
            if (payloads.isEmpty()) {
                try {
                    payload = QUEUE.take();
                } catch (InterruptedException e) {
                    log.warn("从队列获取待发送对象出错", e);
                    Thread.currentThread().interrupt();
                }
            } else {
                try {
                    payload = QUEUE.poll(50, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    log.warn("从队列获取待发送对象出错", e);
                    Thread.currentThread().interrupt();
                }
            }
            if (payload != null) {
                payloads.add(payload);
            } else {
                if (!Collections.isEmpty(payloads)) {
                    // 推送
                    try {
                        this.jClient.sendPushPayloadString(payloads);
                    } finally {
                        payloads.clear();
                    }
                }
            }
            if (payloads.size() == batchThreshold) {
                // 推送
                try {
                    this.jClient.sendPushPayloadString(payloads);
                } finally {
                    payloads.clear();
                }
            }
        }
    }

    @Override
    public void sendSingle(List<SingleNotificationPacket<? extends AbstractNotification>> packets,
            RequestCallback callback) {
        log.info("推送服务推送信息.....");
        if (callback == null) {
            this.singleToBuild(packets, null);
        } else {
            this.sendSingleFuture(packets, callback);
        }
    }

    private void sendSingleFuture(List<SingleNotificationPacket<? extends AbstractNotification>> packets,
            RequestCallback callback) {
        MyFutureTask task = new MyFutureTask();
        try {
            task.start();
            this.singleToBuild(packets, task);
            task.put(new PushPayloadWrapper(null, true));// 最后插入一个元素表示结束
            List<PushResult> pushResultList = task.getPushResultList();
            this.processResult(pushResultList, null, callback);
        } finally {
            task.stop();
        }
    }

    private void singleToBuild(List<SingleNotificationPacket<? extends AbstractNotification>> packets,
            MyFutureTask task) {
        try {
            PushCommand cmd = null;
            for (SingleNotificationPacket<? extends AbstractNotification> packet : packets) {
                if (packet.getCommand() == null) {
                    if (cmd == null) {
                        cmd = new PushCommand();
                    }
                    packet.setCommand(cmd);
                }
                List<String> users = this.getTargetUser(packet.getCommand());
                List<List<String>> userList =
                        Collections.isEmpty(users) ? null : Lists.partition(users, aliasThreshold);// userList决定了单个packet可以分为几条PushPayload
                List<String> tags = packet.getCommand().getTags();
                List<String> tagAndList = packet.getCommand().getTagAndList();
                List<String> tagNotList = packet.getCommand().getTagNotList();

                List<PushPayload.Builder> builders = newBuilder().addPart(packet.getMessage())
                        .builder(packet.getCommand().getTimeToLive(), packet.getSendno());
                new BinancePayload(builders, userList, tags, tagAndList, tagNotList, task).build();
            }
        } finally {
            threadLocal.remove();
            threadLocalMap.remove();
        }
    }

    @Override
    public void sendMultiple(List<MultipleNotificationPacket> packets, RequestCallback callback) {
        if (callback == null) {
            this.multipleToBuild(packets, null);
        } else {
            this.sendMultipleFuture(packets, callback);
        }
    }

    private void sendMultipleFuture(List<MultipleNotificationPacket> packets, RequestCallback callback) {
        MyFutureTask task = new MyFutureTask();
        try {
            task.start();
            this.multipleToBuild(packets, task);
            task.put(new PushPayloadWrapper(null, true));// 最后插入一个元素表示结束
            List<PushResult> pushResultList = task.getPushResultList();
            this.processResult(pushResultList, null, callback);
        } finally {
            task.stop();
        }
    }

    private void multipleToBuild(List<MultipleNotificationPacket> packets, MyFutureTask task) {
        try {
            PushCommand cmd = null;
            for (MultipleNotificationPacket packet : packets) {
                if (packet.getCommand() == null) {
                    if (cmd == null) {
                        cmd = new PushCommand();
                    }
                    packet.setCommand(cmd);
                }
                List<String> users = this.getTargetUser(packet.getCommand());
                List<List<String>> userList =
                        Collections.isEmpty(users) ? null : Lists.partition(users, aliasThreshold);// userList决定了单个packet可以分为几条PushPayload
                List<String> tags = packet.getCommand().getTags();
                List<String> tagAndList = packet.getCommand().getTagAndList();
                List<String> tagNotList = packet.getCommand().getTagNotList();

                List<PushPayload.Builder> builders = newBuilder().addPart(packet.getAndroidMessage())
                        .addPart(packet.getIosMessage()).addPart(packet.getPcMessage()).addPart(packet.getWebMessage())
                        .addPart(packet.getAppMessage())
                        .builder(packet.getCommand().getTimeToLive(), packet.getSendno());
                new BinancePayload(builders, userList, tags, tagAndList, tagNotList, task).build();
            }
        } finally {
            threadLocal.remove();
            threadLocalMap.remove();
        }
    }

    private List<String> getTargetUser(PushCommand cmd) {
        Map<PushCommand, List<String>> m = threadLocal.get();
        if (m == null) {
            m = Maps.newHashMap();
            threadLocal.set(m);
        }
        if (m.containsKey(cmd)) {
            return m.get(cmd);
        }
        List<String> users = Lists.newLinkedList();
        if (CollectionUtils.isNotEmpty(cmd.getUsers())) {
            users.addAll(cmd.getUsers());
        } else {
            String tagetKey = cmd.getTargetUserKey();
            if (StringUtils.isNotBlank(tagetKey)) {
                Map<String, List<String>> map = threadLocalMap.get();
                if (map == null) {
                    map = Maps.newHashMap();
                    threadLocalMap.set(map);
                }
                if (map.containsKey(tagetKey)) {
                    return map.get(tagetKey);
                }
                int count = 100000;
                int i = 0;
                List<String> list = null;
                do {
                    list = this.redisTemplate.opsForList().range(tagetKey, count * i, (i + 1) * count - 1);
                    if (list != null) {
                        users.addAll(list);
                    }
                } while (list != null && list.size() == count && (++i) > 0);

                map.put(tagetKey, users);
            }
        }
        m.put(cmd, users);
        return users;
    }

    private void processResult(List<PushResult> resultList, Exception e, RequestCallback callback) {
        if (resultList == null) {
            resultList = Lists.newArrayList();
        }
        callback.completed(resultList);
    }

    @PostConstruct
    public void init() {
        this.production = "prod".equals(env.getProperty("active"));
        if (this.isUseI18n) {
            PathMatchingResourcePatternResolver pmrpr = new PathMatchingResourcePatternResolver();
            org.springframework.core.io.Resource[] resources = new org.springframework.core.io.Resource[0];
            try {
                resources = pmrpr
                        .getResources(PathMatchingResourcePatternResolver.CLASSPATH_URL_PREFIX + "notification-i18n/*");
            } catch (IOException e) {
                log.error("没有找到国际化资源，请配置国际化资源", e);
                throw new IllegalArgumentException("没有找到国际化资源，请配置国际化资源！");
            }
            for (int i = 0; i < resources.length; i++) {
                org.springframework.core.io.Resource res = resources[i];
                String filename = res.getFilename();
                if (filename.startsWith("notification.")) {
                    filename = org.apache.commons.lang.StringUtils.substringAfter(filename, "notification.");
                }
                String language = org.apache.commons.lang.StringUtils.substringBeforeLast(filename, ".");
                Map<String, String> i18nResource = Maps.newHashMap();
                try (InputStream in = res.getInputStream()) {
                    try (Scanner sc = new Scanner(in)) {
                        while (sc.hasNext()) {
                            String line = sc.nextLine();
                            if (org.apache.commons.lang.StringUtils.isBlank(line)
                                    || org.apache.commons.lang.StringUtils.startsWith(line, "#")) {
                                continue;
                            }
                            String key = org.apache.commons.lang.StringUtils
                                    .trim(org.apache.commons.lang.StringUtils.substringBefore(line, "="));
                            String value = org.apache.commons.lang.StringUtils
                                    .trim(org.apache.commons.lang.StringUtils.substringAfter(line, "="));
                            i18nResource.put(key, value);
                        }
                    }
                } catch (IOException e) {
                    log.error("", e);
                }
                i18nResources.put(language, i18nResource);
            }
        }
        for (int i = 0; i < 30; i++) {
            worker.submit(() -> {
                try {
                    JClient.setNoNeedPushResult();
                    this.sendFromQueue();
                } finally {
                    JClient.removeContext();
                }
            });
        }
    }

    public Builder newBuilder() {
        return new Builder();
    }

    public class Builder {
        private Notification.Builder notificationBuilder;
        private Platform.Builder notificationPlatformBuilder;

        public Builder addPart(AbstractNotification data) {
            if (data == null) {
                return this;
            }
            Map<String, String> extras = createExtras(data);
            if (data instanceof AndroidPayload) {
                this.addAndroid(data, extras);
            } else if (data instanceof IosPayload) {
                this.addIos(data, extras);
            } else if (data instanceof AppPayload) {
                this.addAndroid(data, extras);
                this.addIos(data, extras);
            } else if (data instanceof PcPayload) {
                this.addPc(data, extras);
            } else if (data instanceof WebPayload) {
                this.addWeb(data, extras);
            } else if (data instanceof SimpleNotification) {
                this.addAndroid(data, extras);
                this.addIos(data, extras);
                this.addPc(data, extras);
                this.addWeb(data, extras);
            }
            return this;
        }

        private void addAndroid(AbstractNotification data, Map<String, String> extras) {
            if (jpushAndroidEnabled) {
                ifNotificationBuilderNull();
                if (extras == null) {
                    extras = Maps.newHashMap();
                }
                extras.put(Constants.RESOLVER_KEY, Constants.RESOLVER_VALUE);
                extras.put(Constants.URI_ACTIVITY_KEY,
                        production ? Constants.URI_ACTIVITY_VALUE : Constants.URI_ACTIVITY_VALUE_DEV);
                notificationBuilder.addPlatformNotification(AndroidNotification.newBuilder().setAlert(data.getContent())
                        .setTitle(data.getTitle()).addExtras(extras).build());
                notificationPlatformBuilder.addDeviceType(DeviceType.Android);
                extras.remove(Constants.RESOLVER_KEY);
                extras.remove(Constants.URI_ACTIVITY_KEY);
            }
        }

        private void addIos(AbstractNotification data, Map<String, String> extras) {
            if (jpushIosEnabled) {
                ifNotificationBuilderNull();
                IosAlert iosAlert =
                        IosAlert.newBuilder().setTitleAndBody(data.getTitle(), null, data.getContent()).build();
                IosNotification.Builder builder = IosNotification.newBuilder();
                if(!iosBadgeEnabled){
                    builder.disableBadge();
                }else{
                    if(iosBadgeValue>=0){
                        builder.setBadge(iosBadgeValue);
                    }
                }
                notificationBuilder.addPlatformNotification(
                        builder.setAlert(iosAlert).addExtras(extras).build());
                notificationPlatformBuilder.addDeviceType(DeviceType.IOS);
            }
        }

        private void addPc(AbstractNotification data, Map<String, String> extras) {
            if (jpushPcEnabled) {
                ifNotificationBuilderNull();
                notificationBuilder.addPlatformNotification(PcNotification.newBuilder().setAlert(data.getContent())
                        .setTitle(data.getTitle()).addExtras(extras).build());
                notificationPlatformBuilder.addDeviceType(DeviceType.Windows);
            }
        }

        private void addWeb(AbstractNotification data, Map<String, String> extras) {
            if (jpushWebEnabled) {
                ifNotificationBuilderNull();
                notificationBuilder.addPlatformNotification(WebNotification.newBuilder().setAlert(data.getContent())
                        .setTitle(data.getTitle()).addExtras(extras).build());
                notificationPlatformBuilder.addDeviceType(DeviceType.Web);
            }
        }

        private void ifNotificationBuilderNull() {
            if (notificationBuilder == null) {
                notificationBuilder = Notification.newBuilder();
            }
            if (notificationPlatformBuilder == null) {
                notificationPlatformBuilder = Platform.newBuilder();
            }
        }


        @SuppressWarnings("unchecked")
        private Map<String, String> createExtras(Object body) {
            Map<String, String> extras = null;
            JSONObject jsonObj = (JSONObject) JSONObject.toJSON(body);
            if (body instanceof AbstractNotification) {
                jsonObj.remove("id");
                jsonObj.remove("title");
                jsonObj.remove("content");
                Object payload = jsonObj.remove("payload");
                if (payload != null) {
                    jsonObj.put("others", payload);
                }
            }
            if (!jsonObj.isEmpty()) {
                for (Iterator<String> iter = jsonObj.keySet().iterator(); iter.hasNext();) {
                    String key = iter.next();
                    Object value = jsonObj.get(key);
                    if (value != null) {
                        if (value instanceof Date) {
                            jsonObj.put(key, String.valueOf(((Date) value).getTime()));
                        } else {
                            jsonObj.put(key, value.toString());
                        }
                    }
                }
                extras = jsonObj.toJavaObject(Map.class);
            }
            return extras;
        }

        public List<PushPayload.Builder> builder(long timeToLive, int sendno) {
            List<PushPayload.Builder> pushPayloadBuilders = Lists.newArrayList();
            if (this.notificationBuilder != null) {
                PushPayload.Builder nPayloadBuilder = PushPayload.newBuilder();
                nPayloadBuilder.setOptions(jClient.newOptionsBuilder(timeToLive).setSendno(sendno).build());
                nPayloadBuilder.setPlatform(notificationPlatformBuilder.build());
                nPayloadBuilder.setNotification(notificationBuilder.build());
                pushPayloadBuilders.add(nPayloadBuilder);
            }
            return pushPayloadBuilders;
        }
    }

    private class BinancePayload {

        private List<PushPayload.Builder> builders;
        private MyFutureTask task;
        private List<List<String>> userList;
        private AudienceTarget tags;
        private AudienceTarget tagAnd;
        private AudienceTarget tagNot;

        public BinancePayload(List<PushPayload.Builder> builders, List<List<String>> userList, List<String> tags,
                List<String> tagAndList, List<String> tagNotList, MyFutureTask task) {
            this.builders = builders;
            this.task = task;
            this.userList = userList;

            if (tagNotList == null) {
                tagNotList = Lists.newArrayList();
            } else {
                tagNotList = Lists.newArrayList(tagNotList.iterator());
            }
            tagNotList.add(Constants.PUSH_SWITCH);// 总开关，凡是具有这个标签的都不会收到信息
            this.tags = Collections.isEmpty(tags) ? null : AudienceTarget.tag(tags);
            this.tagNot = AudienceTarget.tag_not(tagNotList);

            if (this.tags == null && Collections.isEmpty(tagAndList) && Collections.isEmpty(this.userList)
                    && this.tagNot != null) {
                this.tagAnd = AudienceTarget.tag_and(Constants.GLOBAL_TAG);
            } else {
                this.tagAnd = Collections.isEmpty(tagAndList) ? null : AudienceTarget.tag_and(tagAndList);
            }
        }

        public void build() {
            if (Collections.isEmpty(builders)) {
                return;
            }
            if (Collections.isEmpty(this.userList) && this.tags == null && this.tagAnd == null && this.tagNot == null) {
                this.all();
            } else {
                this.mix();
            }
        }

        private void all() {
            for (PushPayload.Builder builder : this.builders) {
                builder.setAudience(Audience.all());
                this.toQueue(builder.build());
            }
        }

        private void mix() {
            if (Collections.isEmpty(this.userList)) {
                audienceBuild(null);
            } else {
                for (List<String> users : userList) {
                    Audience.Builder audienceBuilder = Audience.newBuilder();
                    audienceBuilder.addAudienceTarget(AudienceTarget.alias(users));
                    audienceBuild(audienceBuilder);
                }
            }
        }

        private void audienceBuild(Audience.Builder audienceBuilder) {
            if (audienceBuilder == null) {
                audienceBuilder = Audience.newBuilder();
            }
            if (this.tags != null) {
                audienceBuilder.addAudienceTarget(this.tags);
            }
            if (this.tagAnd != null) {
                audienceBuilder.addAudienceTarget(this.tagAnd);
            }
            if (this.tagNot != null) {
                audienceBuilder.addAudienceTarget(this.tagNot);
            }
            Audience audience = audienceBuilder.build();
            for (PushPayload.Builder builder : this.builders) {
                builder.setAudience(audience);
                this.toQueue(builder.build());
            }
        }

        private void toQueue(PushPayload payload) {
            String payloadString = payload.toString();
            List<String> payloadStringList = useI18n(payloadString);
            for (String payloadStr : payloadStringList) {
                try {
                    if (task != null) {
                        task.put(new PushPayloadWrapper(payloadStr, false));
                    } else {
                        QUEUE.put(payloadStr);
                    }
                } catch (InterruptedException e) {
                    log.error("阻塞向队列中放入PushPayload时出现异常：", e);
                    Thread.currentThread().interrupt();
                }
            }
        }

        private List<String> useI18n(String payloadString) {
            List<String> list = Lists.newArrayList();
            if (!isUseI18n) {
                list.add(payloadString);
            } else {
                String[] keys = org.apache.commons.lang.StringUtils.substringsBetween(payloadString, "#{", "}");
                if (keys == null || keys.length == 0) {
                    list.add(payloadString);
                } else {
                    String[] searchList = new String[keys.length];
                    String[] replacementList = new String[keys.length];
                    String[][] values = new String[keys.length][];
                    for (int i = 0; i < keys.length; i++) {
                        String key = keys[i];
                        searchList[i] = "#{" + keys[i] + "}";
                        String value = org.apache.commons.lang.StringUtils.substringAfter(key, ":");
                        if (org.apache.commons.lang.StringUtils.isNotBlank(value)) {
                            values[i] = org.apache.commons.lang.StringUtils.split(value, ",");
                        }
                        keys[i] = org.apache.commons.lang.StringUtils.substringBefore(key, ":");
                    }
                    Set<String> languages = null;
                    try {
                        languages = deviceService.languageTags();
                    } catch (APIConnectionException | APIRequestException e) {
                        languages = Sets.newHashSet(Constants.DEFAULT_LANGUAGE);
                    }
                    // 本次实施国际化是否使用过默认语言
                    String useDefaultLanguagePayload = null;
                    Set<String> includeLanguages = Sets.newHashSet();
                    for (String language : languages) {
                        if (StringUtils.equals(language, Constants.DEFAULT_LANGUAGE)
                                && StringUtils.isNoneBlank(useDefaultLanguagePayload)) {
                            continue;
                        }
                        String tmpLanguage = language;
                        Map<String, String> i18nResource = i18nResources.get(tmpLanguage);
                        if (i18nResource == null) {
                            includeLanguages.add(language);
                            if (StringUtils.isNoneBlank(useDefaultLanguagePayload)) {
                                continue;
                            }
                            tmpLanguage = Constants.DEFAULT_LANGUAGE;
                            i18nResource = i18nResources.get(tmpLanguage);
                        }
                        for (int i = 0; i < keys.length; i++) {
                            String value = i18nResource.get(keys[i]);
                            replacementList[i] = MessageFormat.format(value, values[i]);
                        }
                        String newPayloadString = StringUtils.replaceEach(payloadString, searchList, replacementList);

                        if (StringUtils.equals(tmpLanguage, language)
                                && !StringUtils.equals(language, Constants.DEFAULT_LANGUAGE)) {
                            // 非默认语言
                            newPayloadString =
                                    secondaryProcessingForPayload(newPayloadString, new HashSet<String>(), language);
                            list.add(newPayloadString);
                        } else {
                            // 使用的时默认语言
                            useDefaultLanguagePayload = newPayloadString;
                        }
                    }
                    useDefaultLanguagePayload = secondaryProcessingForPayload(useDefaultLanguagePayload,
                            includeLanguages, Constants.DEFAULT_LANGUAGE);
                    list.add(useDefaultLanguagePayload);
                }
            }
            return list;
        }

        private String secondaryProcessingForPayload(String payloadString, Set<String> includeLanguages,
                String selfLanguage) {
            JSONObject object = JSON.parseObject(payloadString);
            JSONObject audience = object.getJSONObject("audience");
            if (!Collections.isEmpty(includeLanguages)) {
                includeLanguages.add(selfLanguage);
                JSONArray tag = audience.getJSONArray("tag");
                if (tag == null) {
                    tag = new JSONArray();
                    audience.put("tag", tag);
                }
                for (String language : includeLanguages) {
                    tag.add(Constants.LANGUAGE_TAG_PRIX + language);
                }
            } else {
                JSONArray tagAnd = audience.getJSONArray("tag_and");
                if (tagAnd == null) {
                    tagAnd = new JSONArray();
                    audience.put("tag_and", tagAnd);
                }
                tagAnd.add(Constants.LANGUAGE_TAG_PRIX + selfLanguage);
                tagAnd.remove(Constants.GLOBAL_TAG);
            }
            return object.toJSONString();
        }
    }

    private static class PushPayloadWrapper {
        private final String payloadString;
        private final boolean end;

        public PushPayloadWrapper(String payloadString, boolean end) {
            this.payloadString = payloadString;
            this.end = end;
        }

        public String getPayload() {
            return payloadString;
        }

        public boolean getEnd() {
            return this.end;
        }
    }

    private class MyFutureTask {
        private LinkedBlockingQueue<PushPayloadWrapper> queue = Queues.newLinkedBlockingQueue();
        private List<PushResult> collectedPushResult = Lists.newCopyOnWriteArrayList();
        private CountDownLatch c;
        private int workerCount;
        private volatile boolean stop;

        public MyFutureTask() {
            this.workerCount = futureTaskCount;
            if (this.workerCount < 1) {
                this.workerCount = 1;
            }
            this.c = new CountDownLatch(workerCount);
        }

        public void start() {
            for (int i = 0; i < this.workerCount; i++) {
                worker.submit(() -> {
                    collectedPushResult.addAll(this.receiveFromMyQueue());
                    c.countDown();
                });
            }
        }

        public void stop() {
            this.stop = true;
        }

        public void put(PushPayloadWrapper payload) {
            try {
                this.queue.put(payload);
            } catch (InterruptedException e) {
                log.error("阻塞向队列中放入PushPayload时出现异常：", e);
                Thread.currentThread().interrupt();
            }
        }

        public List<PushResult> getPushResultList() {
            try {
                c.await();
            } catch (InterruptedException e) {
                log.error("等待本次所有发送任务完成时出错", e);
                Thread.currentThread().interrupt();
            } catch (Exception e) {
                log.error("等待本次所有发送任务完成时出错", e);
            }
            return this.collectedPushResult;
        }

        private List<PushResult> receiveFromMyQueue() {
            List<PushResult> pushResultList = Lists.newArrayList();
            List<String> payloads = Lists.newArrayList();
            for (;;) {
                PushPayloadWrapper payloadWrapper = null;
                try {
                    payloadWrapper = queue.poll(50, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    log.warn("从队列获取待发送对象出错", e);
                    Thread.currentThread().interrupt();
                }
                if (payloadWrapper != null) {
                    if (payloadWrapper.getEnd()) {
                        log.warn("从队列获取到本次发送任务的最后一个元素，即将结束本次任务");
                        this.stop = true;
                    }
                    if (payloadWrapper.getPayload() != null) {
                        payloads.add(payloadWrapper.getPayload());
                        if (this.stop) {
                            this.toPush(pushResultList, payloads);
                            break;
                        }
                    }
                } else {
                    if (payloads.size() > 0) {
                        // 推送
                        this.toPush(pushResultList, payloads);
                    }
                    if (this.stop) {
                        break;
                    }
                }
                if (payloads.size() == batchThreshold) {
                    // 推送
                    this.toPush(pushResultList, payloads);
                    if (this.stop) {
                        break;
                    }
                }
            }
            return pushResultList;
        }

        private void toPush(List<PushResult> pushResultList, List<String> payloads) {
            try {
                PushResult result = jClient.sendPushPayloadString(payloads);
                pushResultList.add(result);
            } finally {
                payloads.clear();
            }
        }
    }

}
